package ipl;

import java.io.IOException;
import java.util.StringTokenizer;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.*;
import org.apache.hama.HamaConfiguration;
import org.apache.hama.bsp.BSP;
import org.apache.hama.bsp.BSPJob;
import org.apache.hama.bsp.BSPJobClient;
import org.apache.hama.bsp.BSPPeer;
import org.apache.hama.bsp.ClusterStatus;
import org.apache.hama.bsp.FileOutputFormat;
import org.apache.hama.bsp.NullInputFormat;
import org.apache.hama.bsp.TextOutputFormat;
import org.apache.hama.bsp.sync.SyncException;

public class BSPScan {

  private static Path TMP_OUTPUT = new Path("/tmp/pi-"
      + System.currentTimeMillis());

  public static class LocalScan extends
      BSP<LongWritable, Text, LongWritable, Text, IntPair> {
    public static final Log LOG = LogFactory.getLog(BSPScan.class);
    private String masterTask;
    private static final int iterations = 2;

    @Override
    public void bsp(
        BSPPeer<LongWritable, Text, LongWritable, Text, IntPair> peer)
        throws IOException, SyncException, InterruptedException {

      LongWritable key = new LongWritable(-1);
      long sum = 0;
      Text line = new Text();
     // StringBuffer newList = new StringBuffer(512);
      while(peer.readNext(key, line)){
        StringTokenizer itr = new StringTokenizer(line.toString().trim());
        while (itr.hasMoreTokens()) {
                sum += Integer.parseInt(itr.nextToken());
                
        }
      }
      //send p-sum to all
      // key: last line-key
      for(String pn: peer.getAllPeerNames()){
        peer.send(pn, new IntPair(key.get(),sum));
      }
      
      peer.sync();
      
      
    }

    @Override
    public void setup(
        BSPPeer<LongWritable, Text, LongWritable, Text, IntPair> peer)
        throws IOException {
      // Choose one as a master
      //this.masterTask = peer.getPeerName(peer.getNumPeers() / 2);
    }

    @Override
    public void cleanup(
        BSPPeer<LongWritable, Text, LongWritable, Text, IntPair> peer)
        throws IOException {
      if (peer.getPeerName().equals(masterTask)) {
        double pi = 0.0;
        int numPeers = peer.getNumCurrentMessages();
        IntPair received;
        while ((received = peer.getCurrentMessage()) != null) {
// do something
        }

        pi = pi / numPeers;
      //  peer.write(new Text("Estimated value of PI is"), new DoubleWritable(pi));
      }
    }

  }

  static void printOutput(HamaConfiguration conf) throws IOException {
    FileSystem fs = FileSystem.get(conf);
    FileStatus[] files = fs.listStatus(TMP_OUTPUT);
    for (int i = 0; i < files.length; i++) {
      if (files[i].getLen() > 0) {
        FSDataInputStream in = fs.open(files[i].getPath());
        IOUtils.copyBytes(in, System.out, conf, false);
        in.close();
        break;
      }
    }

    fs.delete(TMP_OUTPUT, true);
  }

  public static void main(String[] args) throws InterruptedException,
      IOException, ClassNotFoundException {
    // BSP job configuration
    HamaConfiguration conf = new HamaConfiguration();

    BSPJob bsp = new BSPJob(conf, LocalScan.class);
    // Set the job name
    bsp.setJobName("Pi Estimation Example");
    bsp.setBspClass(LocalScan.class);
    bsp.setInputFormat(NullInputFormat.class);
    bsp.setOutputKeyClass(Text.class);
    bsp.setOutputValueClass(DoubleWritable.class);
    bsp.setOutputFormat(TextOutputFormat.class);
    FileOutputFormat.setOutputPath(bsp, TMP_OUTPUT);

    BSPJobClient jobClient = new BSPJobClient(conf);
    ClusterStatus cluster = jobClient.getClusterStatus(true);

    if (args.length > 0) {
      bsp.setNumBspTask(Integer.parseInt(args[0]));
    } else {
      // Set to maximum
      bsp.setNumBspTask(cluster.getMaxTasks());
    }

    long startTime = System.currentTimeMillis();
    if (bsp.waitForCompletion(true)) {
      printOutput(conf);
      System.out.println("Job Finished in "
          + (System.currentTimeMillis() - startTime) / 1000.0 + " seconds");
    }
  }
}
